package com.ruoyi.flink.wc;

import java.util.List;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoFlatMapFunction;
import org.apache.flink.util.Collector;
import com.ruoyi.flink.util.ParticipleUtil;
import com.ruoyi.flink.wc.entity.Participles;
import com.ruoyi.flink.wc.util.ParticiplesRead;
import com.ruoyi.flink.wc.util.ParticiplesWrite;

public class EventKeyFx {
	
	public static void main(String[] args) throws Exception {
        // 1创建流式环境
		StreamExecutionEnvironment env =  StreamExecutionEnvironment.getExecutionEnvironment();
		// 2创建测试数据源
		DataStreamSource<String> stream = env.fromElements("5月14日，编号为B-001J的C919大飞机从上海浦东机场第4跑道起飞，标志着中国商飞公司即将交付首家用户的首架C919大飞机首次飞行试验圆满完成。",
				"5月15日0-24时，31个省（自治区、直辖市）和新疆生产建设兵团报告新增确诊本土病例140例，新增本土无症状感染者1019例。",
				"在5月16日召开的上海市新冠肺炎疫情防控新闻发布会上，上海市副市长宗明宣布，上海全市16个区当中15个区实现社会面清零。"
				);
	   // 从数据库获取数据源
        String sqlread = "select words, num,timestamp from participles";
        DataStreamSource<Participles> participlesRead = env.addSource(new ParticiplesRead(sqlread));
		// 3. 转换数据格式
		
		 SingleOutputStreamOperator<Tuple2<String, Long>> wordAndOne = stream
	                .flatMap((String eventmsg, Collector<Tuple2<String, Long>> out) -> {
	                	List<String> list = ParticipleUtil.participles(eventmsg);
	                	for (String str : list) {
	                		out.collect(Tuple2.of(str, 1L));
						}
	                }).returns(Types.TUPLE(Types.STRING, Types.LONG));
		// 合并数据
		 SingleOutputStreamOperator<Tuple2<String, Long>> wordAndOneAll =  wordAndOne.connect(participlesRead).flatMap(new CoFlatMapFunction<Tuple2<String, Long>,Participles,Tuple2<String, Long>>(){
			@Override
			public void flatMap1(Tuple2<String, Long> value, Collector<Tuple2<String, Long>> out) throws Exception {
				out.collect(value);
			}
			@Override
			public void flatMap2(Participles value, Collector<Tuple2<String, Long>> out) throws Exception {
			   System.out.println(value.toString());
				out.collect(Tuple2.of(value.words, value.num));
			}
		 });
		 
	    // 4 按照0位单词进行分组
		 KeyedStream<Tuple2<String, Long>, String> wordAndOneKS = wordAndOneAll.keyBy(t -> t.f0);
	   // 5. 求和	
		 SingleOutputStreamOperator<Tuple2<String, Long>> result = wordAndOneKS
	                .sum(1);
	    // 6将数据放入数据库 
		 String sqlWrite = "INSERT INTO participles ( id,words, num,timestamp) VALUES (?,?,?,?)";
		 result.addSink(new ParticiplesWrite(sqlWrite)); 
		 
		// 执行
		 env.execute(); 
		 
		 
	}
}
